Types of streams

stream 有分四種

• Writable：可寫
• Readable：可讀
• Duplex：可讀寫
• Transform：這次不會介紹到它

HTTP server 視角：Readable 與 Writable

HTTP client 視角：Readable 與 Writable

從上述的例子，我們可以得知，所謂的 "Readable" 跟 "Writable"，是根據 "你目前的角色" 來看

client / server 對稱結構

Node.js http 模組的底層就是 stream 跟 net.Socket：

• stream 是管理資料讀寫的抽象層
• net.Socket 則是管理 TCP 連線的抽象層，繼承了 stream.Duplex，可讀寫資料

stream.Writable

https://nodejs.org/api/stream.html#class-streamwritable

stream.Writable 是一個 Base Class + Template Class，它處理所有的 stream 邏輯（buffering、backpressure、events...），但把「實際寫入」的部分留給開發者實作。而這個「實際寫入」的部分，就是 _write 這個 internal method

❌ 錯誤作法

import { Writable } from "stream";

const myWritable = new Writable();
// ❌ 沒有實作 `_write` method 就直接呼叫 `write`
myWritable.write("123"); // Error: The _write() method is not implemented

✅ 正確做法

import { Writable } from "stream";

class MyWritable extends Writable {
  // ✅ 實作 `_write` method
  _write(
    chunk: any,
    encoding: BufferEncoding,
    callback: (error?: Error | null) => void,
  ): void {
    console.log(chunk);
    callback();
  }
}
const myWritable = new MyWritable();
myWritable.write("123"); // <Buffer 31 32 33>

events

從這邊就可以看到 stream.Writable 繼承 EventEmitter 的影子

• writable.on("close")
• writable.on("drain")
• writable.on("error")
• writable.on("finish")
• writable.on("pipe")
• writable.on("unpipe")

methods

• writable.cork
• writable.uncork
• writable.destroy
• writable.write
• writable.end
• writable.setDefaultEncoding

http.OutgoingMessage 繼承鏈

以上方法可以在 Node.js 的 http.ClientRequest 或是 http.ServerResponse 看到，因為其繼承鏈為

為何 http.OutgoingMessage 不是繼承 stream.Writable ? 因為當時 stream.Writable 根本還沒被實作

http.OutgoingMessage 是在 v0.1.17 加入的

stream.Writable 是在 v0.9.4 加入的

但 http.OutgoingMessage 也是實作了 stream.Writable 該有的 methods

http.ClientRequest

import { request } from "http";
const clientRequest = request("http://localhost:5000/");
// ✅ These are all valid methods from stream.Writable
clientRequest.cork();
clientRequest.uncork();
clientRequest.destroy();
clientRequest.write();
clientRequest.end();
clientRequest.setDefaultEncoding();

http.ServerResponse

import http from "http";

const httpServer = http.createServer().listen(5000);
httpServer.on("request", (req, res: http.ServerResponse) => {
  // ✅ These are all valid methods from stream.Writable
  res.cork();
  res.uncork();
  res.destroy();
  res.write();
  res.end();
  res.setDefaultEncoding();
});

properties

先參考即可，下一篇文章會帶到

• writable.closed
• writable.destroyed
• writable.writable
• writable.writableAborted
• writable.writableEnded
• writable.writableFinished
• writable.writableCorked
• writable.errored
• writable.writableHighWaterMark
• writable.writableLength
• writable.writableNeedDrain
• writable.writableObjectMode

internal methods

https://nodejs.org/api/stream.html#implementing-a-writable-stream

這些底線開頭的 internal methods，只有當你在實作 stream.Writable 的時候才會碰到

• writable._write
• writable._writev
• writable._destroy
• writable._final
• writable._construct

❌ 錯誤作法

import { Writable } from "stream";

class MyWritable extends Writable {
  _write(
    chunk: any,
    encoding: BufferEncoding,
    callback: (error?: Error | null) => void,
  ): void {
    console.log(chunk);
    callback();
  }
}
const myWritable = new MyWritable();
myWritable._write("123"); // ❌ instance 直接呼叫 internal methods

✅ 正確做法

import { Writable } from "stream";

class MyWritable extends Writable {
  _write(
    chunk: any,
    encoding: BufferEncoding,
    callback: (error?: Error | null) => void,
  ): void {
    console.log(chunk);
    callback();
  }
}
const myWritable = new MyWritable();
myWritable.write("123"); // ✅ instance 只呼叫 public methods

stream.Readable

https://nodejs.org/api/stream.html#class-streamreadable

同理 stream.Writable，stream.Readable 是一個 Base Class + Template Class，它處理所有的 stream 邏輯（buffering、backpressure、events...），但把「實際讀取」的部分留給開發者實作。而這個「實際讀取」的部分，就是 _read 這個 internal method

❌ 錯誤作法

import { Readable } from "stream";

const myReadable = new Readable();
// ❌ 沒有實作 `_read` method 就直接呼叫 `read`
myReadable.read("123"); // Error: The _read() method is not implemented

✅ 正確做法

import { Readable } from "stream";

class MyReadable extends Readable {
  // ✅ 實作 `_read` method
  _read(size: number): void {
    this.push("123");
    // https://nodejs.org/api/stream.html#readablepushchunk-encoding
    // Passing chunk as null signals the end of the stream (EOF), after which no more data can be written.
    this.push(null);
  }
}

const myReadable = new MyReadable();
myReadable.on("readable", () => {
  const chunk = myReadable.read();
  console.log(chunk); // <Buffer 31 32 33>
});

小插曲：Null-terminated byte strings

在 C 語言的 string 也有 "用 null 當作結尾" 的概念，叫做 Null-terminated byte strings

char greeting[] = "Hello"; // Compiler automatically adds '\0'
// In memory: {'H', 'e', 'l', 'l', 'o', '\0'};

這個概念在 exploit path traversal 的時候，可以當作一個技巧，可參考 PortSwigger - path traversal 的教學

因為 Node.js 底層也是 C，所以我猜測 this.push(null) 這個設計模式也是借鑑 C

events

從這邊就可以看到 stream.Readable 繼承 EventEmitter 的影子

• readable.on("close")
• readable.on("data")
• readable.on("end")
• readable.on("error")
• readable.on("pause")
• readable.on("readable")
• readable.on("resume")

methods

• readable.destroy
• readable.isPaused
• readable.pause
• readable.pipe
• readable.unpipe
• readable.read
• readable.resume
• readable.setEncoding
• readable.unshift
• readable.wrap
• readable.compose
• readable.iterator

http.IncomingMessage 繼承鏈

這些方法都可以在 Node.js 的 http.IncomingMessage 看到，因為其繼承鏈為

As an HTTP client (receive HTTP response)

import http from "http";

const clientRequest = http.request("http://localhost:5000/");
clientRequest.on("response", (response: http.IncomingMessage) => {
  // ✅ These are all valid methods from stream.Readable
  response.destroy();
  response.isPaused();
  response.pause();
  // ......
});

As an HTTP server (receive HTTP request)

import http from "http";

const httpServer = http.createServer().listen(5000);
httpServer.on("request", (req: http.IncomingMessage, res) => {
  // ✅ These are all valid methods from stream.Readable
  req.destroy();
  req.isPaused();
  req.pause();
  // ......
});

properties

先參考即可，下一篇文章會帶到

• readable.closed
• readable.destroyed
• readable.readable
• readable.readableAborted
• readable.readableDidRead
• readable.readableEncoding
• readable.readableEnded
• readable.errored
• readable.readableFlowing
• readable.readableHighWaterMark
• readable.readableLength
• readable.readableObjectMode

internal methods

這些底線開頭的 internal methods，只有當你在實作 stream.Readable 的時候才會碰到

• readable._construct
• readable._destroy
• readable._read

而 readable.push 比較特別，雖然是 public method，但通常也是實作者在 _read 裡面會呼叫的 method

❌ 錯誤作法

import { Readable } from "stream";

class MyReadable extends Readable {
  _read(size: number): void {
    this.push("123");
    // https://nodejs.org/api/stream.html#readablepushchunk-encoding
    // Passing chunk as null signals the end of the stream (EOF), after which no more data can be written.
    this.push(null);
  }
}

const myReadable = new MyReadable();
myReadable._read(); // ❌ instance 直接呼叫 internal methods

✅ 正確做法

import { Readable } from "stream";

class MyReadable extends Readable {
  _read(size: number): void {
    this.push("123");
    // https://nodejs.org/api/stream.html#readablepushchunk-encoding
    // Passing chunk as null signals the end of the stream (EOF), after which no more data can be written.
    this.push(null);
  }
}

const myReadable = new MyReadable();
myReadable.on("readable", () => {
  const chunk = myReadable.read(); // ✅ instance 只呼叫 public methods
  console.log(chunk); // <Buffer 31 32 33>
});

Experimental methods

截止 Node.js v25.2.1，目前這些 methods 都還在 Stability: 1 - Experimental，故不會在本篇深入解說，就當作參考看看

不過，這些 Array-Like methods，在 Node.js MongoDB Driver 也能看到類似的設計模式，我推測是為了提供更好的 DX 吧！

• readable.map
• readable.filter
• readable.forEach
• readable.toArray
• readable.some
• readable.find
• readable.every
• readable.flatMap
• readable.drop
• readable.take
• readable.reduce

stream.Duplex

實作了 stream.Readable 跟 stream.Writable，另外多了 duplex.allowHalfOpen 這個參數，它的意思是

"如果 Readable side 已經結束，那 Writable side 是否要繼續開著？"

聽起來很繞口，我實際舉個實際例子，讓各位了解：

Node.js http.Server，每個 request / response 關聯到的 net.Socket 就是 allowHalfOpen = true，因為通常 server 收到完整的 HTTP request (Readable 結束) 之後，才能決定 HTTP response 是什麼，並且回傳給 client，此時 Writable side 就必須保持開啟

我們可以寫一個 PoC 來驗證 allowHalfOpen = true

import { createServer } from "http";

const httpServer = createServer((req, res) => {
  console.log(req.socket === res.socket); // true
  console.log(req.socket.allowHalfOpen); // true
  console.log(res.socket?.allowHalfOpen); // true
  res.end();
});
httpServer.listen(5000);

Web API 竟然也有 Stream ?!

在校稿的時候，我發現瀏覽器跟 Node.js 都有實作一套 Readable, Writable，名稱分別為

不過 Node.js 在 v16.5.0 也加入了 ReadableStream 跟 WritableStream，對 JavaScript 開發者來說是一大福音，減少學習成本

我在 Transfer-Encoding 這篇文章 也有提到 ReadableStream 的簡易概念，有興趣的夥伴可以參考

小結

這篇文章帶大家入門 stream.Readable, stream.Writable 以及 stream.Duplex

並且也帶入 Node.js http 模組的概念，讓大家從 HTTP 的觀點來理解這些抽象層

下一篇文章，會帶大家深入 stream 的生命週期，以及主要 methods 跟 properties 的使用方法

參考資料

• https://nodejs.org/api/stream.html